package com.hzw.saas.web.admin.job.task;

import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.hzw.saas.api.hpc.IHpcAccountRecordService;
import com.hzw.saas.api.hpc.IHpcFetchService;
import com.hzw.saas.api.hpc.bo.HpcAccountRecordBo;
import com.hzw.saas.common.quartz.model.ScheduleJob;
import com.hzw.saas.common.util.Json;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.PlatformTransactionManager;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;

/**
 * <p>
 *
 * </p>
 *
 * @author sonam
 * @since 2021/3/10 17:05
 */
@Slf4j
@RequiredArgsConstructor
@Service("hpcProviderTask")
public class HpcProviderTask {
    private final PlatformTransactionManager transactionManager;
    @Resource(name = "hpcAccountRecordServiceImpl")
    private IHpcAccountRecordService hpcAccountRecordService;
    @Resource(name = "sshHpcFetchServiceImpl")
    private IHpcFetchService providerService;
    @Resource(name = "executor")
    private Executor executor;

    /**
     * 统计超算中心账号使用机时明细
     */
    @SneakyThrows
    public void startStatistic(ScheduleJob scheduleJob) {
        String params = scheduleJob.getParams();
        /** 参数为线程数量*/
        int threadCount = parseParams(params);
        Queue queue = getQueueData(scheduleJob.getSystemMark());
        CountDownLatch countDownLatch = new CountDownLatch(threadCount);

        List<Task> taskServices = new ArrayList<>();
        for (int i = 0; i < threadCount; i++) {
            Task task = new Task(countDownLatch);
            task.setQueue(queue);
            task.setHpcFetchService(providerService);
            task.setTransactionManager(transactionManager);
            executor.execute(task);
            taskServices.add(task);
        }

        /** 等待所有线程执行完毕，该任务结束*/
        countDownLatch.await();

        /** 检测线程是否有异常抛出*/
        StringBuilder builder = new StringBuilder();
        taskServices.forEach(taskService -> {
            Exception exception = taskService.getException();
            if (Objects.nonNull(exception)) {
                builder.append(taskService.getTaskName()).append(" error: ").append(exception.getMessage()).append("|");
            }
        });

        if (StringUtils.isNotBlank(builder)) {
            throw new Exception(builder.toString());
        }
    }

    /**
     * @param hpcName 超算名称
     * @return
     */
    protected Queue getQueueData(String hpcName) {
        /** 获取待统计的超算中心账号*/
        List<HpcAccountRecordBo> hpcAccountUserBos = hpcAccountRecordService.getListByHpcConfigName(hpcName);

        if(log.isDebugEnabled()) {
            for (HpcAccountRecordBo hpcAccountUserBo : hpcAccountUserBos) {
                log.debug(hpcAccountUserBo.toString());
            }
        }

        /** 将数据加队列*/
        Queue<HpcAccountRecordBo> queue = new ConcurrentLinkedQueue();
        queue.addAll(hpcAccountUserBos);
        return queue;
    }

    protected int parseParams(String params) {
        int threadCount;
        try {
            Map<String, String> map = Json.getObjectMapper().readValue(params, Map.class);
            threadCount = Integer.parseInt(map.get("threadCount"));
        } catch (Exception e) {
            e.printStackTrace();
            threadCount = 8;
        }
        return threadCount;
    }

}

